package org.xiaoyu.rpc.remoting.codec;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.DefaultMaxBytesRecvByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.xiaoyu.rpc.common.constant.RpcError;
import org.xiaoyu.rpc.common.exception.RpcException;
import org.xiaoyu.rpc.common.exception.SerializeException;
import org.xiaoyu.rpc.common.util.Bytes;
import org.xiaoyu.rpc.remoting.packet.*;
import org.xiaoyu.rpc.seralization.Serializer;
import org.xiaoyu.rpc.seralization.support.SerializerAlgorithm;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;

/**
 * Created By Have
 * 2021/7/28 21:42
 */
@Slf4j
public class Codec {
    private static final int HEADER_LENGTH = 14;
    private static final short MAGIC_NUMBER = (short) 0xacac;
    private static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC_NUMBER)[0];
    private static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC_NUMBER)[1];
    private static final int MAX_LENGTH = 10485760; // 10MB


    public void encode(Channel ch, Packet packet, ByteBuf out) {
        byte[] bytes = new byte[0];
        if (packet.getSerializerType() == SerializerAlgorithm.FASTJSON.getType()) {
            bytes = SerializerAlgorithm.FASTJSON.getSerializer().serializer(packet);
        } else if (packet.getSerializerType() == SerializerAlgorithm.KRYO.getType()) {
            bytes = SerializerAlgorithm.KRYO.getSerializer().serializer(packet);
        } else if (packet.getSerializerType() == SerializerAlgorithm.HESSIAN2.getType()) {
            bytes = SerializerAlgorithm.HESSIAN2.getSerializer().serializer(packet);
        } else if (packet.getSerializerType() == SerializerAlgorithm.PROTOBUF.getType()) {
            bytes = SerializerAlgorithm.PROTOBUF.getSerializer().serializer(packet);
        } else {
            log.error("不存在的序列化器");
            throw new SerializeException("序列化器不存在");
        }
        if (bytes.length >= MAX_LENGTH) {
            throw new RpcException("Packet size must be less than " + MAX_LENGTH);
        }

        out.writeByte(MAGIC_HIGH);
        out.writeByte(MAGIC_LOW);
        out.writeInt(packet.getType());
        out.writeInt(packet.getSerializerType());
        out.writeInt(bytes.length);
        out.writeBytes(bytes);

        // For Test
//        StringBuilder sb = new StringBuilder();
//        byte[] bytes1 = new byte[out.readableBytes()];
//        out.readBytes(bytes1);
//        String s = bytesToHexString(bytes1);
//        sb.append(s);
//        out.resetReaderIndex();
//        System.out.println(sb);
        out.writeByte(10); // TCP测试用
    }

    public static String bytesToHexString(byte[] src){
        StringBuilder stringBuilder = new StringBuilder("");
        if (src == null || src.length <= 0) {
            return null;
        }
        for (int i = 0; i < src.length; i++) {
            int v = src[i] & 0xFF;
            String hv = Integer.toHexString(v);
            if (hv.length() < 2) {
                stringBuilder.append(0);
            }
            stringBuilder.append(hv);
        }
        return stringBuilder.toString();
    }
    public Object decode(Channel channel, ByteBuf buf) throws IOException {
        int readable = buf.readableBytes();
        byte[] header = getHeader(buf);
        // 这里对应的Dubbo telnet的解包情况，后续待实现
        if (readable > 0 && MAGIC_HIGH != header[0]
        || readable > 1 && header[1] != MAGIC_LOW) { // 与魔数对不上的情况
            int length = header.length;
            if (header.length < readable) {
                header = Bytes.copyOf(header, readable);
                buf.readBytes(header, length, readable - length); // 填充数据到bytes
            }
            for (int i = 1; i < header.length - 1; i++) {
                if (header[i] == MAGIC_HIGH && header[i+1] == MAGIC_LOW) {
                    buf.readerIndex(buf.readerIndex() - header.length + i); // 移动到包开始处, 也就是魔数的起点，上一个包的结尾
                    header = Bytes.copyOf(header, i); // 现在的header存的是上一个包的数据
                    break;
                }
                return decode(channel, buf, readable, header); // 粘包的情况
            }
        }

        if (readable < HEADER_LENGTH) {
            return DecodeResult.NEED_MORE_INPUT;
        }
        // 当前包的开端是正确的请求头
        int length = Bytes.bytes2int(header,10); // data length

        int tt = length + HEADER_LENGTH;
        if (readable < tt) {
            return DecodeResult.NEED_MORE_INPUT;
        }

        return decodeBody(channel, header, buf);
    }

    private Object decodeBody(Channel channel, byte[] header, ByteBuf buf) {
        int packetType = Bytes.bytes2int(header, 2);
        int serializerType = Bytes.bytes2int(header, 6);
        int length = Bytes.bytes2int(header,10); // data length

        Class<?> packageClass;
        if (packetType == PacketType.REQUEST_PACK) {
            packageClass = Invocation.class;
        } else if (packetType == PacketType.RESPONSE_PACK) {
            packageClass = AppResponse.class;
        } else if (packetType == PacketType.HEARTBEAT_REQ_PACK) {
            packageClass = HeartbeatRequest.class;
        } else {
            log.error("无法识别的包类型:[{}]", packetType);
            throw new RpcException(RpcError.NOT_SUPPORTED_PACKET);
        }

        Serializer serializer;
        if (serializerType == SerializerAlgorithm.FASTJSON.getType()) {
            serializer = SerializerAlgorithm.FASTJSON.getSerializer();
        } else if (serializerType == SerializerAlgorithm.KRYO.getType())  {
            serializer = SerializerAlgorithm.KRYO.getSerializer();
        } else if (serializerType == SerializerAlgorithm.HESSIAN2.getType()) {
            serializer = SerializerAlgorithm.HESSIAN2.getSerializer();
        } else if (serializerType == SerializerAlgorithm.PROTOBUF.getType()) {
            serializer = SerializerAlgorithm.PROTOBUF.getSerializer();
        } else {
            log.error("无法识别的序列化器类型:[{}]", serializerType);
            throw new RpcException(RpcError.NOT_SUPPORTED_SERIALIZER);
        }

        byte[] bytes = new byte[length];
        buf.readBytes(bytes);
        return serializer.deserializer(packageClass, bytes);
    }

    public Object decode(Channel channel, ByteBuf buf, int readable, byte[] message) throws IOException {
        if (message == null || message.length == 0) {
            return DecodeResult.NEED_MORE_INPUT;
        }
        return null;
    }

    public byte[] getHeader(ByteBuf buf) {
        int readable = buf.readableBytes();
        byte[] bytes = new byte[Math.min(HEADER_LENGTH, readable)];
        buf.readBytes(bytes);
        return bytes;
    }

    public static void main(String[] args) {
        ByteBuf buffer = PooledByteBufAllocator.DEFAULT.buffer();
        byte[] bytes = {1, 0, 0, 0, 1, 2, 3};
        byte[] bytes1 = Bytes.copyOf(bytes, 3);
        System.out.println(Arrays.toString(bytes1));

        System.out.println(Bytes.bytes2int(bytes, 1));

        ByteBuf byteBuf = buffer.writeBytes(bytes);
        System.out.println(byteBuf.readerIndex() + ":" + byteBuf.writerIndex());
        ByteBuf byteBuf1 = byteBuf.readBytes(bytes);
        System.out.println(byteBuf.readerIndex() + ":" + byteBuf.writerIndex());
    }

    public enum DecodeResult {
        NEED_MORE_INPUT, SKIP_SOME_INPUT;
    }
}
